package com.heima.safe.aliyun.mq;

import com.alibaba.fastjson.JSON;
import com.heima.model.media.pojos.WmNews;
import com.heima.safe.aliyun.verify.VerifyImage;
import com.heima.safe.aliyun.verify.VerifyText;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Component
public class VerifyQueueListener {
    @Autowired
    private VerifyText verifyText;
    @Autowired
    private VerifyImage verifyImage;
    @Autowired
    private KafkaTemplate kafkaTemplate;


    /**
     * VerifyQueue队列监听
     */
    @KafkaListener(topics = "VerifyQueue",groupId = "verifygroup")
    public void verifyQueue(ConsumerRecord<?,?>record)throws Exception{
        //1.获取数据，包含文本、图片、原始数据、监听结果的MQ("backQueue","verifyArticle")
           //其中，文本内容的key为text，value为List<String>
           //其中，图片的key为images，value为List<String>
           //其中，原始数据的key为attach，value为WmNews
        Map<String,Object>dataMap = JSON.parseObject(record.value().toString(), Map.class);
        //2.安全风控,获取检测结果verifyResult：整合所有检测结果
        HashMap<String, Object> verifyResult = new HashMap<>();

        //2.1获取内容，及检测
        List<String> contents = (List<String>)dataMap.get("text");
        //文本内容检测
        if (contents!=null&&contents.size()>0){
            //textResult：是文本内容的检测结果
            //join方法：以逗号分割拼接成string
            Map<String, String> textResult = verifyText.verify(StringUtils.join(contents, ","));
            //putAll方法：将一个已有Map中的数据压入另一个Map中，且去重。
            if (textResult!=null){
                verifyResult.putAll(textResult);
            }

        }

        //2.2.获取图片，及检测
        List<String> images = (List<String>)dataMap.get("images");
        //图片内容检测
        if (images!=null&&images.size()>0){
            Map<String, String> imageResult = verifyImage.verify(images);
            if (imageResult!=null){
                verifyResult.putAll(imageResult);
            }


        }

        //3.原始数据
        verifyResult.put("attach",dataMap.get("attach"));
        //4.监听结果的队列名字（此队列名称由wemedia服务提供，已经发送过来了，只需获取）
        String backQueue =  dataMap.get("backQueue").toString();
        System.out.println("========================");

        //将检测的的结果集发送回MQ
        kafkaTemplate.send(backQueue,JSON.toJSONString(verifyResult));

    }


}
